线程池拒绝策略

接口：RejectedExecutionHandler
默认使用； private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
JDK已有实现子类：
CallerRunsPolicy 调用者线程自己执行
AbortPolicy 拒绝抛出异常 带有异常信息 "Task " + r.toString() +" rejected from " + e.toString() 
DiscardPolicy 直接抛弃，无异常
DiscardOldestPolicy 弃旧迎新，将任务队列的旧的线程剔除一个，然后执行当前任务

其他实现
seata 包实现 内部类：取出旧的任务，调用者线程执行旧任务，新任务先尝试放入任务队列，放不进则提交给线程池决定执行方式
dubbo AbortPolicyWithReport实现：拒绝任务前打印日志和堆栈信息，
netty NewThreadRunsPolicy实现：新建线程对象执行被线程池拒绝的任务，可能会逐渐占用更多的内存和CPU资源
tomcat 实现： RejectHandler  拒绝抛异常，和jdk自带的拒绝就是少了拒绝信息


自定义实现 RejectedExecutionHandler 接口即可 例如：

public class MyRejectPolicy implements RejectedExecutionHandler{
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        //Sender是我的Runnable类，里面有message字段
        //你自己的拒绝逻辑
    }
}

jdk内置实现
/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */

package java.util.concurrent;

/**
 * A handler for tasks that cannot be executed by a {@link ThreadPoolExecutor}.
 *
 * @since 1.5
 * @author Doug Lea
 拒绝策略
 */
public interface RejectedExecutionHandler {

    /**
     * Method that may be invoked by a {@link ThreadPoolExecutor} when
     * {@link ThreadPoolExecutor#execute execute} cannot accept a
     * task.  This may occur when no more threads or queue slots are
     * available because their bounds would be exceeded, or upon
     * shutdown of the Executor.
     *
     * <p>In the absence of other alternatives, the method may throw
     * an unchecked {@link RejectedExecutionException}, which will be
     * propagated to the caller of {@code execute}.
     *
     * @param r the runnable task requested to be executed 请求执行的任务
     * @param executor the executor attempting to execute this task 任务执行器
     * @throws RejectedExecutionException if there is no remedy

     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

    /** 直接拒绝 ，抛出拒绝异常
     * A handler for rejected tasks that throws a
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }
    /**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the {@code execute} method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     在调用者的线程中执行任务的策略
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }
     /** 抛弃策略，不做任何处理
     * A handler for rejected tasks that silently discards the
     * rejected task.
     */
    public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardPolicy}.
         */
        public DiscardPolicy() { }

        /**
         * Does nothing, which has the effect of discarding task r.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

     /** 抛弃旧的任务，执行新的任务
     * A handler for rejected tasks that discards the oldest unhandled
     * request and then retries {@code execute}, unless the executor
     * is shut down, in which case the task is discarded.
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardOldestPolicy} for the given executor.
         */
        public DiscardOldestPolicy() { }

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

package io.seata.common.thread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public final class RejectedPolicies {
    public RejectedPolicies() {
    }

    public static RejectedExecutionHandler runsOldestTaskPolicy() {
        return new RejectedExecutionHandler() {
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (!executor.isShutdown()) {
                    BlockingQueue<Runnable> workQueue = executor.getQueue();
                    Runnable firstWork = (Runnable)workQueue.poll();
                    boolean newTaskAdd = workQueue.offer(r);
                    if (firstWork != null) {
                        firstWork.run();//拿出队列老的任务，如果不为空，那么就当前线程执行旧任务
                    }

                    if (!newTaskAdd) {//如果新任务没有加到工作队列中
                        executor.execute(r);//就提交给线程池去觉得执行
                    }

                }
            }
        };
    }
}
//dubbo包 org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport
拒绝，添加报告，日志典范
public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy{
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        String msg = String.format("Thread pool is EXHAUSTED!" +
                " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: "
                + "%d)," +
                " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
            threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(),
            e.getLargestPoolSize(),
            e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
            url.getProtocol(), url.getIp(), url.getPort());
        logger.warn(msg);
        dumpJStack();//另开一个独立的线程dump出堆栈信息Dubbo_JStack.log文件
        throw new RejectedExecutionException(msg);
    }
}
netty包 org.jboss.netty.handler.execution.MemoryAwareThreadPoolExecutor.NewThreadRunsPolicy
//新建线程执行 被拒绝的任务，会逐渐占用全部的内存cpu等资源
 private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {
        private NewThreadRunsPolicy() {
        }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            try {
                Thread t = new Thread(r, "Temporary task executor");
                t.start();
            } catch (Throwable var4) {
                throw new RejectedExecutionException("Failed to start a new thread", var4);
            }
        }
    }

//拒绝抛异常，和jdk自带的拒绝就是少了拒绝信息
private static class RejectHandler implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r,
                java.util.concurrent.ThreadPoolExecutor executor) {
            throw new RejectedExecutionException();
        }

    }